package com.dahuan.stream;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.TopologyBuilder;

import java.util.Properties;

public class Application {

    public static void main(String[] args) {

        //TODO 创建拓扑对象
        TopologyBuilder builder = new TopologyBuilder();
        //TODO 创建配置文件
        Properties props = new Properties();

        props.put( "application.id","Application" );
        props.put( "bootstrap.servers","huan01:9092" );

        //TODO 构建拓扑结构
        builder.addSource( "SOURCE","first" )
                .addProcessor("PROCESSOR", new ProcessorSupplier() {
                    @Override
                    public Processor get() {
                        return new LogProcessor();
                    }
                },"SOURCE"  )
                .addSink( "SINK", "second","PROCESSOR");

        //TODO 创建流对象
        KafkaStreams kafkaStreams = new KafkaStreams( builder, props );

        //TODO 开始运行
        kafkaStreams.start();

    }
}
